package com.simple.dispatcher.service;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.annotation.PostConstruct;
import javax.sql.DataSource;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.json.simple.JSONArray;
import org.json.simple.JSONObject;
import org.json.simple.parser.JSONParser;
import org.json.simple.parser.ParseException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import weibo4j.Weibo;
import weibo4j.WeiboException;
import weibo4j.http.AccessToken;

@Service
public class DispatcherService {
	
	protected final Log log = LogFactory.getLog(getClass());
	
	@Autowired
	private DataSource dataSource;
	
	private ConcurrentHashMap<Long, String> localFilters = new ConcurrentHashMap<Long, String>();
	
	private ExecutorService matchFilterThreadPool = Executors.newSingleThreadExecutor();
	
	private ExecutorService sendWeiboThreadPool = Executors.newFixedThreadPool(5);
	
	@PostConstruct
	public void init() {
		
		localFilters = new ConcurrentHashMap<Long, String>();
		
		try {
			Connection conn = dataSource.getConnection();
			String sql = "SELECT * FROM job_filter";
			PreparedStatement stmt = conn.prepareStatement(sql);
			ResultSet rs = stmt.executeQuery();
			while(rs.next()){
				long userId = rs.getLong("user_id");
				String filter = rs.getString("filter");
				localFilters.put(userId, filter);
			}
		} catch (SQLException e) {
			e.printStackTrace();
		}
		
		log.info("load filters: " + localFilters.size());
	}
	
	public boolean match(String filter, String msgMeta) {
		//String metaData = "{\"industry\": [\"计算机\", \"互联网\"], \"jobtitle\": [\"软件研发\", \"软件测试\"], \"location\": [\"上海\", \"北京\"], \"salary\": 5000, \"company\": \"网易\", \"type\": [\"全职\", \"兼职\"]}";
		//String filter = "{\"industry\": [\"互联网\", \"计算机\"], \"jobtitle\": [\"软件研发\"], \"location\": [\"上海\", \"北京\", \"西安\"], \"salary\": {\"value\": 5000, \"op\": \"ge\"}}";
		
		JSONParser parser = new JSONParser();
		try {
			JSONObject data = (JSONObject) parser.parse(msgMeta);
			JSONObject userFilter = (JSONObject) parser.parse(filter);
			Iterator<Entry<String, Object>> itr = (Iterator<Entry<String, Object>>) userFilter.entrySet().iterator();
			while(itr.hasNext()) {
				Entry<String, Object> filterItem = itr.next();
				Object filterItemValue = filterItem.getValue();
				if(filterItemValue instanceof String || filterItemValue instanceof Number) {
					//基本数据类型
					String dataItemValue = data.get(filterItem.getKey()).toString();
					if(!filterItemValue.equals(dataItemValue)) {
						return false;
					}
				} else if (filterItemValue instanceof JSONObject) {
					if(((JSONObject)filterItemValue).containsKey("value") && ((JSONObject)filterItemValue).containsKey("op")) {
						//带操作符的条件(salary)
						 Long value1 = (Long)data.get(filterItem.getKey());
						 Long value2 = (Long)((JSONObject)filterItemValue).get("value");
						 String op = (String)((JSONObject)filterItemValue).get("op");
						 if(!compareLong(value1, value2, op)) {
							 return false;
						 }
					}
				} else if (filterItemValue instanceof JSONArray) {
					//数组
					Iterator arrayItr = ((JSONArray) filterItemValue).iterator();
					boolean isMatch = false;
					while(arrayItr.hasNext()) {
						Object filValue = arrayItr.next();
						//基本数据类型
						String dataItemValue = data.get(filterItem.getKey()).toString();
						if(dataItemValue.contains(filValue.toString())) {
							isMatch = true;
						}
					}
					if(!isMatch) {
						return false;
					}
				}
			}
		} catch (ParseException e) {
			e.printStackTrace();
		}
		return true;
	}
	
	private boolean compareLong(long value1, long value2, String op) {
		if("gt".equals(op)) {
			 return value1 > value2;
		} else if("ge".equals(op)) {
			 return value1 >= value2;
		} else if("lt".equals(op)) {
			 return value1 < value2;
		} else if("le".equals(op)) {
			 return value1 <= value2;
		} else if("eq".equals(op)) {
			 return value1 == value2;
		}
		return false;
	}
	
	public void send(final long weiboId, final long userId) {
		
		Runnable task = new Runnable() {
			
			@Override
			public void run() {
				try {
					Token token = getUserAccessToken(userId);
					Weibo weibo = new Weibo();
					weibo.setToken(new AccessToken(token.getToken(), token.getTokenSecret()));
					
					weibo.createFavorite(weiboId);
				} catch (WeiboException e) {
					e.printStackTrace();
				}
			}
		};
		
		sendWeiboThreadPool.submit(task);
	}
	
	public void dispatch(final long weiboId, final String msgMeta) {
		
		Runnable task = new Runnable() {
			
			@Override
			public void run() {
				
				for(Entry<Long, String> entry : localFilters.entrySet()) {
					long userId = entry.getKey();
					String filter = entry.getValue();
					if(match(filter, msgMeta)){
						send(weiboId, userId);
					}
				}
				
			}
		};
		
		matchFilterThreadPool.submit(task);
	}
	
	public Token getUserAccessToken(long userId) {
		try{
			String sql = "SELECT * FROM user WHERE user_id=?";
			Connection conn = dataSource.getConnection();
			PreparedStatement stmt = conn.prepareStatement(sql);
			stmt.setLong(1, userId);
			ResultSet rs = stmt.executeQuery();
			if(rs.next()){
				Token token = new Token();
				token.setUid(rs.getLong("user_id"));
				token.setToken(rs.getString("oauth_access_token"));
				token.setTokenSecret(rs.getString("oauth_access_token_secret"));
				return token;
			}
		}catch(SQLException e){
			e.printStackTrace();
		}
		return null;
	}

	public void addUserFilter(long userId, String filter) {
		try {
			Connection conn = dataSource.getConnection();
			PreparedStatement selectStmt = conn.prepareStatement("SELECT count(*) FROM job_filter WHERE user_id = ?");
			selectStmt.setLong(1, userId);
			ResultSet rs = selectStmt.executeQuery();
			if(rs.next()) {
				int count = rs.getInt(1);
				if(count == 0) {
					PreparedStatement stmt = conn.prepareStatement("INSERT INTO job_filter (user_id, filter) VALUES (?, ?)");
					stmt.setLong(1, userId);
					stmt.setString(2, filter);
					stmt.executeUpdate();
					stmt.close();
				} else {
					PreparedStatement stmt = conn.prepareStatement("UPDATE job_filter SET filter = ? WHERE user_id = ?");
					stmt.setString(1, filter);
					stmt.setLong(2, userId);
					stmt.executeUpdate();
					stmt.close();
				}
			}
			conn.close();
			localFilters.put(userId, filter);
			log.info("subscribe success: userId = " + userId + " size: " + localFilters.size());
		} catch (SQLException e) {
			e.printStackTrace();
		}
	}
	
	public void deleteUserFilter(long userId) {
		try{
			String sql = "DELETE FROM job_filter WHERE user_id=?";
			Connection conn = dataSource.getConnection();
			PreparedStatement stmt = conn.prepareStatement(sql);
			stmt.setLong(1, userId);
			stmt.executeUpdate();
			stmt.close();
			conn.close();
		
		} catch(SQLException e) {
			e.printStackTrace();
		}
		
		localFilters.remove(userId);
	}
}
